use std::{ops::Deref, time::Duration};

use futures::{future::BoxFuture, stream::StreamExt, FutureExt};
use serde::{
    de::{self, Deserializer},
    Deserialize,
};

use crate::{
    bson::{Bson, Deserializer as BsonDeserializer, Document},
    bson_util,
    error::Result,
    options::{FindOptions, Hint, InsertManyOptions, UpdateOptions},
    test::util::{CommandEvent, EventClient},
    Collection,
};

pub(super) trait TestOperation {
    /// The command names to monitor as part of this test.
    fn command_names(&self) -> &[&str];

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>>;
}

pub(super) struct AnyTestOperation {
    operation: Box<dyn TestOperation>,
}

impl<'de> Deserialize<'de> for AnyTestOperation {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
        #[derive(Deserialize)]
        struct OperationDefinition {
            name: String,
            arguments: Bson,
        }

        let definition = OperationDefinition::deserialize(deserializer)?;
        let boxed_op = match definition.name.as_str() {
            "insertOne" => InsertOne::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "insertMany" => InsertMany::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "updateOne" => UpdateOne::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "updateMany" => UpdateMany::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "deleteMany" => DeleteMany::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "deleteOne" => DeleteOne::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            "find" => Find::deserialize(BsonDeserializer::new(definition.arguments))
                .map(|op| Box::new(op) as Box<dyn TestOperation>),
            _ => unimplemented!(),
        }
        .map_err(|e| de::Error::custom(format!("{}", e)))?;

        Ok(AnyTestOperation {
            operation: boxed_op,
        })
    }
}

impl Deref for AnyTestOperation {
    type Target = Box<dyn TestOperation>;

    fn deref(&self) -> &Box<dyn TestOperation> {
        &self.operation
    }
}

#[derive(Deserialize)]
pub(super) struct DeleteMany {
    filter: Document,
}

impl TestOperation for DeleteMany {
    fn command_names(&self) -> &[&str] {
        &["delete"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            collection
                .delete_many(self.filter.clone(), None)
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

#[derive(Deserialize)]
pub(super) struct DeleteOne {
    filter: Document,
}

impl TestOperation for DeleteOne {
    fn command_names(&self) -> &[&str] {
        &["delete"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            collection
                .delete_one(self.filter.clone(), None)
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

#[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct Find {
    filter: Option<Document>,
    // `FindOptions` cannot be embedded directly because serde doesn't support combining `flatten`
    // and `deny_unknown_fields`, so its fields are replicated here.
    #[serde(default)]
    sort: Option<Document>,
    #[serde(default)]
    skip: Option<u64>,
    #[serde(default, rename = "batchSize")]
    batch_size: Option<i64>,
    #[serde(default)]
    limit: Option<i64>,
    #[serde(default)]
    comment: Option<String>,
    #[serde(default)]
    hint: Option<Hint>,
    #[serde(
        rename = "maxTimeMS",
        deserialize_with = "bson_util::deserialize_duration_option_from_u64_millis",
        default
    )]
    max_time: Option<Duration>,
    #[serde(default)]
    min: Option<Document>,
    #[serde(default)]
    max: Option<Document>,
    #[serde(rename = "returnKey", default)]
    return_key: Option<bool>,
    #[serde(rename = "showRecordId", default)]
    show_record_id: Option<bool>,
}

impl TestOperation for Find {
    fn command_names(&self) -> &[&str] {
        &["find", "getMore"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            // `FindOptions` is constructed without the use of `..Default::default()` to enforce at
            // compile-time that any new fields added there need to be considered here.
            let options = FindOptions {
                sort: self.sort.clone(),
                skip: self.skip,
                batch_size: self.batch_size.map(|i| i as u32),
                limit: self.limit,
                comment: self.comment.clone(),
                hint: self.hint.clone(),
                max_time: self.max_time,
                min: self.min.clone(),
                max: self.max.clone(),
                return_key: self.return_key,
                show_record_id: self.show_record_id,
                allow_disk_use: None,
                allow_partial_results: None,
                cursor_type: None,
                max_await_time: None,
                max_scan: None,
                no_cursor_timeout: None,
                projection: None,
                read_concern: None,
                selection_criteria: None,
                collation: None,
            };

            let mut cursor = collection.find(self.filter.clone(), options).await?;

            while let Some(result) = cursor.next().await {
                result?;
            }
            Ok(())
        }
        .boxed()
    }
}

#[derive(Debug, Deserialize)]
pub(super) struct InsertMany {
    documents: Vec<Document>,
    #[serde(default)]
    options: Option<InsertManyOptions>,
}

impl TestOperation for InsertMany {
    fn command_names(&self) -> &[&str] {
        &["insert"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            collection
                .insert_many(self.documents.clone(), self.options.clone())
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

#[derive(Debug, Deserialize)]
pub(super) struct InsertOne {
    document: Document,
}

impl TestOperation for InsertOne {
    fn command_names(&self) -> &[&str] {
        &["insert"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            collection
                .insert_one(self.document.clone(), None)
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

#[derive(Debug, Deserialize)]
pub(super) struct UpdateMany {
    filter: Document,
    update: Document,
}

impl TestOperation for UpdateMany {
    fn command_names(&self) -> &[&str] {
        &["update"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            collection
                .update_many(self.filter.clone(), self.update.clone(), None)
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

#[derive(Debug, Deserialize)]
pub(super) struct UpdateOne {
    filter: Document,
    update: Document,
    #[serde(default)]
    upsert: Option<bool>,
}

impl TestOperation for UpdateOne {
    fn command_names(&self) -> &[&str] {
        &["update"]
    }

    fn execute(&self, collection: Collection<Document>) -> BoxFuture<Result<()>> {
        async move {
            let options = self.upsert.map(|b| UpdateOptions {
                upsert: Some(b),
                ..Default::default()
            });
            collection
                .update_one(self.filter.clone(), self.update.clone(), options)
                .await
                .map(|_| ())
        }
        .boxed()
    }
}

impl EventClient {
    pub(super) async fn run_operation_with_events(
        &self,
        operation: AnyTestOperation,
        database_name: &str,
        collection_name: &str,
    ) -> Vec<CommandEvent> {
        let _: Result<_> = operation
            .execute(self.database(database_name).collection(collection_name))
            .await;
        self.get_command_events(operation.command_names())
    }
}
